package com.wfg.bigdata.spark.sparksql.udf;

import org.apache.spark.sql.Encoder;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.expressions.Aggregator;

import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author wufagang
 * @description
 * @date 2024年05月13日 16:43
 * // 自定义函数
 * //      1. 创建自定义的【公共】类
 * //      2. 继承 org.apache.spark.sql.expressions.Aggregator
 * //      3. 设定泛型
 * //          IN : 输入数据类型
 * //          BUFF : 缓冲区的数据类型
 * //          OUT : 输出数据类型
 * //     4. 重写方法 （ 4（计算） + 2(状态)）
 */
public class CityRemarkUDAF extends Aggregator<String, CityRemarkBuffer,String> {
    /**
     * 缓冲区的初始化操作
     */
    @Override
    public CityRemarkBuffer zero() {
        return new CityRemarkBuffer();
    }

    /**
     * 将输入的地区和缓冲区数据聚合
     */
    @Override
    public CityRemarkBuffer reduce(CityRemarkBuffer b, String a) {
//        if(Objects.isNull(b)) b = new CityRemarkBuffer();
        b.setTotal(b.getTotal()+1);
        b.getCityMap().merge(a,1l, (v, k)->{
            v = v+1;
            return v;
        });
        return b;
    }

    /**
     * 合并缓冲区的数据
     */
    @Override
    public CityRemarkBuffer merge(CityRemarkBuffer b1, CityRemarkBuffer b2) {
        b1.setTotal(b1.getTotal()+b2.getTotal());
        Map<String, Long> cityMap = b1.getCityMap();
        Map<String, Long> cityMap1 = b2.getCityMap();
        cityMap1.forEach((k,v) -> {
            cityMap.merge(k,v,(v1,k1)->{
                return v1+v;
            });
        });
        return b1;
    }

    String form = "地区:%s占比:%s";
    /**
     * 将统计的结果生成字符串信息
     */
    @Override
    public String finish(CityRemarkBuffer reduction) {
        Map<String, Long> cityMap = reduction.getCityMap();
        Long total = reduction.getTotal();
        StringBuffer sb = new StringBuffer();
        System.out.println(reduction.getCityMap());
        cityMap.forEach((k,v) ->{
            sb.append(String.format(form, k, calculateRatio(v,total)));
        });
        return sb.toString();
    }

    public String calculateRatio(long numerator, long denominator) {
        if (denominator == 0) {
            throw new IllegalArgumentException("Denominator cannot be zero.");
        }
        double ratio = (double) numerator / denominator;
        return String.format("%.2f%%", ratio * 100);
    }

    @Override
    public Encoder<CityRemarkBuffer> bufferEncoder() {
        return Encoders.bean(CityRemarkBuffer.class);
    }

    @Override
    public Encoder<String> outputEncoder() {
        return Encoders.STRING();
    }


}
